package com.liyueheng.transmitToMySQL

import com.liyueheng.util.{ConfigLoader, MySQLConnect, SparkConf}
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions.{col, when}

object ExportDwsToMySQL {

  def exportDwsToMySQL(): Unit = {
    // 创建 SparkSession
    val spark: SparkSession = SparkConf.createSparkSession("ExportDwsToMySQL")

    // 获取 DWS 数据库名称
    val dwsDb = ConfigLoader.getString("databases.dws")

    // 切换到 dws 数据库
    spark.sql(s"use $dwsDb")

    // 获取所有表
    //    val tables = spark.catalog.listTables(dwsDb).collect().map(_.name)
    val tables = spark.catalog.listTables("gmall_dws").collect().map(_.name)

    // 遍历所有表并写入 MySQL
    tables.foreach { tableName =>
      try {

        val df = spark.table(s"$dwsDb.$tableName")

        df.write
          .mode(SaveMode.Overwrite)
          .format("jdbc")
          .option("url", ConfigLoader.getString("db.url"))
          .option("dbtable", tableName)
          .option("user", ConfigLoader.getString("db.username"))
          .option("password", ConfigLoader.getString("db.password"))
          .option("driver", ConfigLoader.getString("db.driver"))
          .save()

        println(s"✅ 表 [$tableName] 写入 MySQL 成功")

      } catch {
        case e: Exception =>
          println(s"❌ 表 [$tableName] 写入 MySQL 失败: ${e.getMessage}")
      }
    }

    // 停止 Spark
    SparkConf.stopSparkSession(spark)
  }

  def main(args: Array[String]): Unit = {
    exportDwsToMySQL()
  }
}
